package tbdp.tas

import java.sql.{DriverManager, Connection}
import java.text.{DecimalFormat, SimpleDateFormat}
import java.util.concurrent.ConcurrentHashMap
import java.util.{Date, HashMap}

import org.apache.commons.lang.StringUtils
import org.apache.hadoop.hbase.client.{Result, HTableInterface, Put}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig}
import tbdp.tas.hbase.{HBaseConnection, HBaseConnectionFactory}
import tbdp.tas.model.MinsLine
import tbdp.tas.util.db.{DBUtil, DataBaseType}

/**
  * Created by xbkaishui on 16/9/5.
  */
object Utils {

  val cache = new ConcurrentHashMap[String, KafkaProducer[String, String]]()

  /**
    * 获取mysql 链接 可以通过 java -Dpasswd=xxx 方式注入链接信息
    *
    * @return
    */

  def getMysqlConn(): Connection = {
    val passwd = sys.props.getOrElse("passwd", "tbdp123&*()")
    val db = sys.props.getOrElse("db", "tbdp")
    val url = sys.props.getOrElse("url", "192.168.13.44")
    val dbType = DataBaseType.MySql
    val jdbc = s"jdbc:MySQL://$url:3306/$db?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false"
    val conn = DBUtil.getConnection(dbType, jdbc, db, passwd)
    conn
  }


  def isBk(marketType: String): Boolean = {
    StringUtils.equalsIgnoreCase(marketType, "9") || StringUtils.equalsIgnoreCase(marketType, "10")
  }

  def getHbaseTable(table: String): HTableInterface = {
    val zkNode = sys.props.getOrElse("hbase_znode", "/hbase-unsecure");
    val zkAddr = sys.props.getOrElse("hbase_zk", "zj-hz-bn01-hadoop-02,zj-hz-bn01-spark-03,zj-hz-bn01-spark-02,zj-hz-bn01-spark-01,zj-hz-bn01-hadoop-01");
    val conn = new HBaseConnection(zkAddr, zkNode)
    val hbaseConnector = HBaseConnectionFactory.instance().getHbaseConnector(conn)
    //    println("hbase table " + table)
    hbaseConnector.getTable(table)
  }


  def getKafkaProducer(brokers: String): KafkaProducer[String, String] = {
    if (cache.get(brokers) != null) {
      return cache.get(brokers)
    } else {
      val props = new HashMap[String, Object]()
      props.put(ProducerConfig.BATCH_SIZE_CONFIG, "500")
      //      props.put(ProducerConfig.ACKS_CONFIG, "all")
      props.put(ProducerConfig.ACKS_CONFIG, "1")

      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer")
      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer")

      val producer = new KafkaProducer[String, String](props)
      cache.putIfAbsent(brokers, producer)
      producer
    }
  }

  def getDate(): String = {
    val dateSdf = new SimpleDateFormat("yyyyMMdd")
    val date = new Date()
    dateSdf.format(date)
  }
  def getDateTime(): String = {
    val dateSdf = new SimpleDateFormat("yyyyMMdd HH:mm:ss")
    val date = new Date()
    dateSdf.format(date)
  }

  /***
    * 判断当前行情是否结束
    * @return
    */
  def hqEnd(): Boolean = {
    val tm = getCurHHMM()
    tm > "16:30"
  }

  def getCurHHMM(): String = {
    val dateSdf = new SimpleDateFormat("HH:mm")
    val date = new Date()
    val timestemp = dateSdf.format(date)
    timestemp
  }

  def initHbaseTable(table: String): HTableInterface = {
    val zkNode = sys.props.getOrElse("hbase_znode", "/hbase-unsecure");
    val zkAddr = sys.props.getOrElse("hbase_zk", "zj-hz-bn01-hadoop-02,zj-hz-bn01-spark-03,zj-hz-bn01-spark-02,zj-hz-bn01-spark-01,zj-hz-bn01-hadoop-01");
    val conn = new HBaseConnection(zkAddr, zkNode)
    val hbaseConnector = HBaseConnectionFactory.instance().getHbaseConnector(conn)
    println("hbase table " + table)
    hbaseConnector.getTable(table)
  }

  def formatNumber(num: Double): String = {
    val df = new DecimalFormat("###.##")
    df.format(num)
  }

  def formatNumber(num: Double, precise: Int): Double = {
    (s"%.$precise" + "f").format(num).toDouble
  }

  def getHiveConn(): Connection = {
    val driverName = "org.apache.hive.jdbc.HiveDriver"
    Class.forName(driverName);
    val hiveConn = sys.props.getOrElse("hive_conn", "jdbc:hive2://192.168.13.31:10000/tbdp");
    val con = DriverManager.getConnection(hiveConn, "", "");
    con
  }

  def roundTo2Decimals(d: Double): Double = {
    val df2 = new DecimalFormat("###")
    df2.setMaximumFractionDigits(2)
    java.lang.Double.valueOf(df2.format(d))
  }

  def sendMinsTopValToHbase(hbaseTable: HTableInterface, st: MinsLine): Unit = {
    val rowkey = st.marketType + "_" + st.sCode
    val row = new Put(Bytes.toBytes(rowkey))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("SCode"), Bytes.toBytes(st.sCode));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("MarketType"), Bytes.toBytes(st.marketType))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("ATrade"), Bytes.toBytes(st.aTrade.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Vol"), Bytes.toBytes(st.vol.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Amt"), Bytes.toBytes(Utils.formatNumber(st.amt).toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AOpen"), Bytes.toBytes(st.aOpen.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AHigh"), Bytes.toBytes(st.aHigh.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("ALow"), Bytes.toBytes(st.aLow.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AClose"), Bytes.toBytes(st.aClose.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("YClose"), Bytes.toBytes(st.yClose.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("IsTrade"), Bytes.toBytes("1"))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("IsExp"), Bytes.toBytes("1"))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("CodeType"), Bytes.toBytes(st.codeType))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBVol"), Bytes.toBytes(st.agencyBVol.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBAmt"), Bytes.toBytes(Utils.formatNumber(st.agencyBAmt).toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllBVol"), Bytes.toBytes(st.allBVol.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllBAmt"), Bytes.toBytes(Utils.formatNumber(st.allBAmt).toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySVol"), Bytes.toBytes(st.agencySVol.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySAmt"), Bytes.toBytes(Utils.formatNumber(st.agencySAmt).toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSVol"), Bytes.toBytes(st.allSVol.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSAmt"), Bytes.toBytes(Utils.formatNumber(st.allSAmt).toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBSco"), Bytes.toBytes(st.agencyBSco.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySSco"), Bytes.toBytes(st.agencySSco.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("PriTeamBSco"), Bytes.toBytes(st.priTeamBSco.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("PriTeamSSco"), Bytes.toBytes(st.priTeamSSco.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("PublicBSco"), Bytes.toBytes(st.publicBSco.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("PublicSSco"), Bytes.toBytes(st.publicSSco.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSco"), Bytes.toBytes(st.allSco.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Avg5AllSco"), Bytes.toBytes(st.avg5AllSco.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency1BSAmt"), Bytes.toBytes(st.agencyBSAmt1D.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency3BSAmt"), Bytes.toBytes(st.agencyBSAmt3D.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency5BSAmt"), Bytes.toBytes(st.agencyBSAmt5D.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency7BSAmt"), Bytes.toBytes(st.agencyBSAmt7D.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBVolRate"), Bytes.toBytes(st.agencyBVolRate.toString));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllBVolRate"), Bytes.toBytes(st.allBVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySVolRate"), Bytes.toBytes(st.agencySVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSVolRate"), Bytes.toBytes(st.allSVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency1BSVolRate"), Bytes.toBytes(st.agency1BSVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency3BSVolRate"), Bytes.toBytes(st.agency3BSVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency5BSVolRate"), Bytes.toBytes(st.agency5BSVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency10BSVolRate"), Bytes.toBytes(st.agency10BSVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency20BSVolRate"), Bytes.toBytes(st.agency20BSVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllVolRate"), Bytes.toBytes(st.aLLVolRate.toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("All1BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all1BSAmt).toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("All3BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all3BSAmt).toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("All5BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all5BSAmt).toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("All10BSAmt"), Bytes.toBytes(Utils.formatNumber(st.all10BSAmt).toString))
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("TDate"), Bytes.toBytes(st.tDate));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("TTime"), Bytes.toBytes(st.TTime));
    row.addColumn(Bytes.toBytes("cf1"),
      Bytes.toBytes("TFlag"), Bytes.toBytes(st.tflag))
    hbaseTable.put(row)
  }

  /**
    * convert to minsline
    *
    * @param result
    * @param interface
    * @return
    */
  def convert(result: Result, interface: HTableInterface): MinsLine = {
    val min = new MinsLine
    val sCode = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("SCode"))
    val marketType = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("MarketType"))
    val codeType = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("CodeType"))
    val aTrade = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("ATrade"))
    val yClose = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("YClose"))
    val tDate = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("TDate"))
    val tFlag = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("TFlag"))

    val agencyBVol = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBVol"))
    min.agencyBVol = Bytes.toString(agencyBVol).trim.toLong
    val agencyBAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBAmt"))
    min.agencyBAmt = Bytes.toString(agencyBAmt).trim.toDouble
    val allBVol = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllBVol"))
    min.allBVol = Bytes.toString(allBVol).trim.toLong
    val allBAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllBAmt"))
    min.allBAmt = Bytes.toString(allBAmt).trim.toDouble
    val agencySVol = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySVol"))
    min.agencySVol = Bytes.toString(agencySVol).trim.toLong

    val agencySAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySAmt"))
    min.agencySAmt = Bytes.toString(agencySAmt).trim.toDouble

    val allSVol = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSVol"))
    min.allSVol = Bytes.toString(allSVol).trim.toLong

    val allSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSAmt"))
    min.allSAmt = Bytes.toString(allSAmt).trim.toDouble
    val agencyBSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBSco"))
    min.agencyBSco = Bytes.toString(agencyBSco).trim.toInt
    val agencySSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySSco"))
    min.agencySSco = Bytes.toString(agencySSco).trim.toInt
    val priTeamBSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("PriTeamBSco"))
    min.priTeamBSco = Bytes.toString(priTeamBSco).trim.toInt

    val priTeamSSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("PriTeamSSco"))
    min.priTeamSSco = Bytes.toString(priTeamSSco).trim.toInt

    val publicBSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("PublicBSco"))
    min.publicBSco = Bytes.toString(publicBSco).trim.toInt

    val publicSSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("PublicSSco"))
    min.publicSSco = Bytes.toString(publicSSco).trim.toInt
    val allSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSco"))
    min.allSco = Bytes.toString(allSco).trim.toInt

    val avg5AllSco = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Avg5AllSco"))
    min.avg5AllSco = Bytes.toString(avg5AllSco).trim.toInt

    val agency1BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency1BSAmt"))
    min.agencyBSAmt1D = Bytes.toString(agency1BSAmt).trim.toDouble
    val agency3BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency3BSAmt"))
    min.agencyBSAmt3D = Bytes.toString(agency3BSAmt).trim.toDouble
    val agency5BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency5BSAmt"))
    min.agencyBSAmt5D = Bytes.toString(agency5BSAmt).trim.toDouble
    val agency7BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency7BSAmt"))
    min.agencyBSAmt7D = Bytes.toString(agency7BSAmt).trim.toDouble

    val agencyBVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencyBVolRate"))
    min.agencyBVolRate = Bytes.toString(agencyBVolRate).trim.toDouble

    val allBVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllBVolRate"))
    min.allBVolRate = Bytes.toString(allBVolRate).trim.toDouble

    val agencySVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AgencySVolRate"))
    min.agencySVolRate = Bytes.toString(agencySVolRate).trim.toDouble

    val allSVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllSVolRate"))
    min.allSVolRate = Bytes.toString(allSVolRate).trim.toDouble

    val agency1BSVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency1BSVolRate"))
    min.agency1BSVolRate = Bytes.toString(agency1BSVolRate).trim.toDouble

    val agency3BSVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency3BSVolRate"))
    min.agency3BSVolRate = Bytes.toString(agency3BSVolRate).trim.toDouble

    val agency5BSVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency5BSVolRate"))
    min.agency5BSVolRate = Bytes.toString(agency5BSVolRate).trim.toDouble

    val agency10BSVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency10BSVolRate"))
    min.agency10BSVolRate = Bytes.toString(agency10BSVolRate).trim.toDouble

    val agency20BSVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("Agency20BSVolRate"))
    min.agency20BSVolRate = Bytes.toString(agency20BSVolRate).trim.toDouble

    val allVolRate = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("AllVolRate"))
    min.aLLVolRate = Bytes.toString(allVolRate).trim.toDouble

    val all1BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("All1BSAmt"))
    min.all1BSAmt = Bytes.toString(all1BSAmt).trim.toDouble
    val all3BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("All3BSAmt"))
    min.all3BSAmt = Bytes.toString(all3BSAmt).trim.toDouble

    val all5BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("All5BSAmt"))
    min.all5BSAmt = Bytes.toString(all5BSAmt).trim.toDouble

    val all10BSAmt = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("All10BSAmt"))
    min.all10BSAmt = Bytes.toString(all10BSAmt).trim.toDouble

    val tTime = result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("TTime"))
    min.TTime = Bytes.toString(tTime).trim

    min.minsCnt = Bytes.toString(result.getValue(Bytes.toBytes("cf1"),
      Bytes.toBytes("MinsCnt"))).trim.toInt

    min.sCode = Bytes.toString(sCode)
    min.marketType = Bytes.toString(marketType)
    min.codeType = Bytes.toString(codeType)
    min.tDate = Bytes.toString(tDate)
    min.tflag = Bytes.toString(tFlag)
    min.aTrade = Bytes.toString(aTrade).trim.toDouble
    min.yClose = Bytes.toString(yClose).trim.toDouble
    min
  }


  def main(args: Array[String]) {
    println(formatNumber(653333507.0064377))
    println(formatNumber(0,2))
  }

}
